/*
 *    Copyright (c) 2018-2025, lengleng All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * Redistributions of source code must retain the above copyright notice,
 * this list of conditions and the following disclaimer.
 * Redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the following disclaimer in the
 * documentation and/or other materials provided with the distribution.
 * Neither the name of the pie4cloud.com developer nor the names of its
 * contributors may be used to endorse or promote products derived from
 * this software without specific prior written permission.
 * Author: lengleng (wangiegie@gmail.com)
 */
package com.pie4cloud.pie.bus.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.nacos.shaded.com.google.gson.Gson;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;

import com.pie4cloud.pie.bus.api.dto.ConsumerDto;
import com.pie4cloud.pie.bus.api.entity.DataKafConsumer;
import com.pie4cloud.pie.bus.mapper.DataKafConsumerMapper;
import com.pie4cloud.pie.bus.service.KafkaManageService;
import com.pie4cloud.pie.bus.service.monitor.Monitor;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * kafka消费者管理
 *
 * @author pie code generator
 * @date 2021-12-28 10:59:58
 */
@Service
@RequiredArgsConstructor
public class KafkaManageServiceImpl extends ServiceImpl<DataKafConsumerMapper, DataKafConsumer> implements KafkaManageService {
	private final DataKafConsumerMapper dataKafConsumerMapper;
	@Value(value = "${spring.kafka.bootstrap-servers}")
	private String bootstrap;
	@Autowired
	private KafkaTemplate kafkaTemplate;
	public IPage<DataKafConsumer> pageAndMonitor(Page page, DataKafConsumer dataKafConsumer){
		IPage<DataKafConsumer> iPage=super.page(page, Wrappers.query(dataKafConsumer));
		List<DataKafConsumer> dataKafConsumerList=iPage.getRecords();
		if(null!=dataKafConsumerList&&dataKafConsumerList.size()>0){
			CountDownLatch countDown = new CountDownLatch(dataKafConsumerList.size());
			for(DataKafConsumer consumer:dataKafConsumerList){
				Thread thread=new Thread(()->{
					try {
						Monitor monitor=new Monitor(bootstrap,consumer.getTopic(),consumer.getKafGroup());
						monitor.monitor(false);
						consumer.setLastTotalLag(monitor.getLastTotalLag());
					} catch (Exception e) {
						e.printStackTrace();
					}
					countDown.countDown();
				});
				thread.start();
			}
			try {
				countDown.await();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}

		}
		return iPage.setRecords(dataKafConsumerList);

	}
	@Transactional
	public DataKafConsumer saveAndSend(DataKafConsumer dataKafConsumer) {
		try {
			ConsumerDto consumerDto=new ConsumerDto();
			BeanUtil.copyProperties(dataKafConsumer,consumerDto);
			Date date=new Date();
			String time=String.valueOf(date.getTime());
			consumerDto.setTime(time);
			this.send(consumerDto);
			dataKafConsumer.setConsumptionTime(date);
			super.save(dataKafConsumer);
		} catch (Exception e) {
			return null;
		}
		return dataKafConsumer;
	}
	public DataKafConsumer editAndSend(DataKafConsumer dataKafConsumer) {
		try {
			ConsumerDto consumerDto=new ConsumerDto();
			BeanUtil.copyProperties(dataKafConsumer,consumerDto);
			Date date=new Date();
			String time=String.valueOf(date.getTime());
			consumerDto.setTime(time);
			DataKafConsumer dataKafConsumerOld=super.getById(dataKafConsumer.getConsId());
			if(null==dataKafConsumer.getConsId()||null==dataKafConsumerOld){
				this.saveAndSend(dataKafConsumer);
			}else {
				if(!(dataKafConsumerOld.getKafGroup()+"_"+dataKafConsumerOld.getTopic()).equals(dataKafConsumer.getKafGroup()+"_"+dataKafConsumer.getTopic())){
					kafkaTemplate.send("stop", new Gson().toJson(consumerDto));
				}
				this.send(consumerDto);
				dataKafConsumer.setConsumptionTime(date);
				super.updateById(dataKafConsumer);
			}

		} catch (Exception e) {
			return dataKafConsumer;
		}
		return dataKafConsumer;
	}
	public boolean delAndSend(Long consId) {
		try {
			DataKafConsumer dataKafConsumer=super.getById(consId);
			ConsumerDto consumerDto=new ConsumerDto();
			BeanUtil.copyProperties(dataKafConsumer,consumerDto);
			Date date=new Date();
			String time=String.valueOf(date.getTime());
			consumerDto.setTime(time);
			kafkaTemplate.send("stop", new Gson().toJson(consumerDto));

			super.removeById(dataKafConsumer.getConsId());
		} catch (Exception e) {
			return false;
		}
		return true;
	}

	public int deleteByTopicAndGroup(String topic , String group){
		return dataKafConsumerMapper.deleteByTopicAndGroup(topic,group);
	}
	public void send(ConsumerDto consumerDto){
		kafkaTemplate.send("stop", new Gson().toJson(consumerDto));
		if(null!=consumerDto.getTriggerStatus()&&1==consumerDto.getTriggerStatus()){
			for(int i=0;i<consumerDto.getWorkNum();i++){
				kafkaTemplate.send("bus", new Gson().toJson(consumerDto));
			}
		}
	}

	public List<DataKafConsumer> selectByTriggerStatus(){
         return dataKafConsumerMapper.selectByTriggerStatus();
	}
	public void resetConsumer(){
		List<DataKafConsumer> consumers=this.selectByTriggerStatus();
        if(null!=consumers&&consumers.size()>0){
        	for(DataKafConsumer dataKafConsumer:consumers){
				ConsumerDto consumerDto=new ConsumerDto();
				BeanUtil.copyProperties(dataKafConsumer,consumerDto);
				Date date=new Date();
				String time=String.valueOf(date.getTime());
				consumerDto.setTime(time);
				this.send(consumerDto);
			}
		}
	}
}
